﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using NetMicro.Cache;
using NetMicro.Core.Ioc.ServiceLocation;
using NetMicro.EventBus.Extensions;
using NetMicro.EventBus.Messages;
using NetMicro.EventBus.Persistences;
using Microsoft.Extensions.DependencyInjection;
using NetMicro.Core.Extensions;

namespace NetMicro.EventBus.BackgroundServices
{
    /// <summary>
    /// 重试服务
    /// </summary>
    public class MessageRetryServiceProvider : IMessageRetryServiceProvider
    {
        private IOptions<EventBusOptions> Options { get; }
        private IRedisCache RedisCache { get; }

        private IMessageSerializer MessageSerializer { get; }

        private IEventHandlerInvoker EventHandlerInvoker { get; }

        private IEventHandlerFindProvider EventHandlerFindProvider { get; }

        private ILogger<MessageRetryServiceProvider> Logger { get; }

        private IEventErrorSendDeadLetter EventErrorSendDeadLetter { get; }
        private IRetryMessageStorage MessageStorage { get; }

        public MessageRetryServiceProvider(IOptions<EventBusOptions> options,
            IRedisCache redisCache,
            IMessageSerializer messageSerializer,
            IEventHandlerInvoker eventHandlerInvoker,
            IEventHandlerFindProvider eventHandlerFindProvider,
            ILogger<MessageRetryServiceProvider> logger,
            IEventErrorSendDeadLetter eventErrorSendDeadLetter,
            IRetryMessageStorage messageStorage)
        {
            Options = options;
            RedisCache = redisCache;
            MessageSerializer = messageSerializer;
            EventHandlerInvoker = eventHandlerInvoker;
            EventHandlerFindProvider = eventHandlerFindProvider;
            Logger = logger;
            EventErrorSendDeadLetter = eventErrorSendDeadLetter;
            MessageStorage = messageStorage;
        }


        public Task Execute(CancellationToken cancellationToken)
        {
            Task.Factory.StartNew(async () =>
               {
                   await HandlerAsync(cancellationToken);
               }, TaskCreationOptions.LongRunning);
            return Task.CompletedTask;
        }


        public async Task HandlerAsync(CancellationToken cancellationToken)
        {
            while (true)
            {
                try
                {
                    IEnumerable<MessageStorageTransfer> retryMessage = await MessageStorage.GetMessagesAsync();
                    if (!retryMessage.Any())
                    {
                        await Task.Delay(TimeSpan.FromSeconds(1));
                        continue;
                    }
                    foreach (MessageStorageTransfer item in retryMessage)
                    {
                        if (item.DelayAt > TimeSpan.Zero.GetTimestampSeconds())
                            continue;

                        string key = $"{EventBusConsts.EventBusTopic}:{item.Id}";
                        bool flag = await RedisCache.SetAsync(key, item.Id, 60);
                        if (flag)
                        {
                            MessageTransfer transfer = MessageSerializer.ConvertMessageTransfer(item);
                            try
                            {
                                await MessageStorage.LockAsync(item.Id);
                                if (transfer != null)
                                {

                                    item.RetryCount++;
                                    EventHandlerDescriptor descriptor = EventHandlerFindProvider.GetByName(transfer.EventHandlerName);
                                    if (descriptor is null)
                                    {
                                        Logger.LogError($"[EventBus] not found of event handler: {transfer.EventHandlerName}, but receive msg: {transfer.MsgBody}.");
                                        continue;
                                    }
                                    DateTime now = DateTime.Now;
                                    transfer.Items ??= new Dictionary<string, string>();
                                    MessageTransfer messageTransfer = new MessageTransfer
                                    {
                                        MsgId = transfer.MsgId,
                                        Version = transfer.Version ?? Options.Value.Version,
                                        SendAt = transfer.SendAt,
                                        ExcuteAt = now,
                                        EventHandlerName = transfer.EventHandlerName,
                                        EventName = transfer.EventName,
                                        Items = transfer.Items,
                                        MsgBody = transfer.MsgBody
                                    };
                                    // 执行事件消费
                                    await EventHandlerInvoker.InvokeAsync(messageTransfer, descriptor).ConfigureAwait(false);
                                    await MessageStorage.DeleteMessageAsync(item.Id);

                                }
                            }
                            catch (Exception ex)
                            {
                                Logger.LogError(ex, "普通事件重试异常");
                                TimeSpan? timespan = transfer?.GetRetry(item.RetryCount);
                                if (timespan == null)
                                {
                                    if (transfer == null)
                                        continue;
                                    transfer.RetryCount = item.RetryCount;
                                    using (IServiceScope scope = ServiceLocator.Current.CreateScope())
                                    {
                                        var tran = scope.ServiceProvider.GetService<ITransactionContext>();
                                        await EventErrorSendDeadLetter.OnSendDeadLetterAsync(transfer, tran.Transaction, cancellationToken);
                                        await MessageStorage.DeleteMessageAsync(item.Id, tran.Transaction);
                                        tran.Commit();
                                    }
                                }
                                else
                                {
                                    await MessageStorage.NextAsync(item.Id, item.RetryCount, timespan ?? TimeSpan.Zero);
                                }
                            }
                        }
                        await RedisCache.DelAsync(key: key);
                    }
                }
                catch (Exception ex)
                {
                    Logger.LogError(ex, "普通事件重试异常");
                    await Task.Delay(500);
                }
            }

        }
    }
}
